package com.ubxtech.mqtt.config;


import com.ubxtech.core.utils.SpringUtils;
import com.ubxtech.mqtt.bo.MqttTopicBO;
import com.ubxtech.mqtt.handler.AbstractReceiveMessageHandler;
import com.ubxtech.mqtt.handler.ReceiveMessageDefaultHandler;
import com.ubxtech.mqtt.properties.MqttProperties;
import com.ubxtech.mqtt.service.EmqTopicManager;
import com.ubxtech.mqtt.service.MqttTopicService;
import com.ubxtech.mqtt.service.impl.EmqTopicManagerImpl;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * https://www.cnblogs.com/songyaru/p/13596660.html
 * https://blog.csdn.net/cuiyaonan2000/article/details/108263617
 * @author Victor.Xiao
 * @date 2021-09-29 17:38
 */
@Configuration
@Slf4j
@ConditionalOnProperty(prefix = MqttProperties.PREFIX, name = "enabled", havingValue = "true", matchIfMissing = true)
public class MqttInboundConfiguration {

    @Value("${spring.mqtt.inbound.username}")
    private String username;

    @Value("${spring.mqtt.inbound.password}")
    private String password;

    @Value("#{'${spring.mqtt.url}'.split(',')}")
    private String[] serverUrls;

    @Value("${spring.mqtt.inbound.clientId}")
    private String clientId;
    /** 是否固定clientId,默认不固定,1,true,false,0 */
    @Value("${spring.mqtt.inbound.clientIdFixed:}")
    private String clientIdFixed;

    @Value("#{'${spring.mqtt.inbound.topics}'.split(',')}")
    private String[] topics;


    private MqttPahoMessageDrivenChannelAdapter adapter;

    @Autowired(required = false)
    private MqttTopicService mqttTopicService;


    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        final DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();

        final MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        //若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true
        mqttConnectOptions.setCleanSession(false);
        mqttConnectOptions.setServerURIs(this.serverUrls);
        mqttConnectOptions.setUserName(this.username);
        mqttConnectOptions.setPassword(this.password.toCharArray());
        //配置最大不确定接收消息数量，默认值10，qos!=0 时生效 在mqtt消息量比较大的情况下将值设大
        mqttConnectOptions.setMaxInflight(MqttConstant.MAX_IN_FLIGHT);
        //配置超时时长，默认30000毫秒
//        mqttConnectOptions.setConnectionTimeout(MqttConstant.CONNECTION_TIMEOUT);
        //防止 ERROR o.e.p.c.mqttv3.internal.ClientState - Timed out as no activity  错误
        mqttConnectOptions.setConnectionTimeout(0);
        //定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待
        mqttConnectOptions.setKeepAliveInterval(MqttConstant.KEEP_ALIVE_TIME);
        mqttConnectOptions.setAutomaticReconnect(true);
        log.info("启动mqtt连接 :{}", mqttConnectOptions);

        factory.setConnectionOptions(mqttConnectOptions);
        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    /**
     * Inbound入站，消费者的消息处理器,接受处理消息
     * @author Victor.Xiao
     * @date 2021-09-29 17:39
     */
    @Bean
    @DependsOn("mqttTopicService")
    public MessageProducer inbound(MqttPahoClientFactory mqttPahoClientFactory ) {
        adapter = new MqttPahoMessageDrivenChannelAdapter(MqttBeanHolderUtil.appendClientId(clientId + "_subscribe",clientIdFixed),mqttPahoClientFactory, topics);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setCompletionTimeout(30000);
        adapter.setQos(MqttConstant.QOS_1);
        adapter.setOutputChannel(mqttInputChannel());


        if (mqttTopicService == null) {
            try {
                mqttTopicService = SpringUtils.getBean("mqttTopicService",MqttTopicService.class);
            } catch (Exception e) {
                log.error("初始化mqttTopicService失败");
            }
        }
        if (mqttTopicService != null) {
            //服务启动后，将所有设备的主题进行监听
            List<MqttTopicBO> mqttShadowTopics = mqttTopicService.getInitTopics(MqttConstant.DEFAULT_CHANNEL_NAME);
            Set<String> topics = new HashSet<>();
            mqttShadowTopics.forEach(mqttShadowTopic -> {
                if (!topics.contains(mqttShadowTopic.getSubTopic())) {
                    adapter.addTopic(mqttShadowTopic.getSubTopic(), 1);
                }
                topics.add(mqttShadowTopic.getSubTopic());
            });
        }
        MqttBeanHolderUtil.putAdapter(MqttConstant.DEFAULT_CHANNEL_NAME,adapter);

        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel") // 绑定消费者
    public MessageHandler handler(AbstractReceiveMessageHandler abstractReceiveMessageHandler) {
        //ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于接收消息信息的channel。
        return abstractReceiveMessageHandler;
    }



    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * Outbound出站，发布者发送的消息通道，发送消息
     * @author Victor.Xiao
     * @date 2021-09-29 17:39
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound(@Autowired final MqttPahoClientFactory mqttPahoClientFactory) {
        final MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(MqttBeanHolderUtil.appendClientId(clientId + "_publish", clientIdFixed), mqttPahoClientFactory);
        //设置异步发送，默认是false(发送时阻塞)
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(MqttConstant.QOS_1);
//        messageHandler.setDefaultTopic(this.topics);
        messageHandler.setDefaultRetained(false);
        messageHandler.setAsyncEvents(false);
        MqttBeanHolderUtil.putMessageHandler(MqttConstant.DEFAULT_CHANNEL_NAME, messageHandler);
        return messageHandler;
    }


    @Bean
    @ConditionalOnMissingBean
    public AbstractReceiveMessageHandler abstractReceiveMessageHandler() {
        return new ReceiveMessageDefaultHandler();
    }

    @Bean
    @ConditionalOnMissingBean
    public EmqTopicManager emqTopicManager() {
        return new EmqTopicManagerImpl();
    }

}